package org.diy.traffic.sinks;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.HashMap;
import java.util.logging.Level;

import org.diy.traffic.sinks.TrafficSink.TrafficSinkException;
import org.diy.traffic.sources.DiyFluSource;
import org.diy.traffic.sources.TrafficSource;
import org.diy.traffic.sources.TrafficSource.TrafficSourceException;
import org.diy.traffic.updates.TrafficUpdate;
import org.diy.traffic.updates.TrafficUpdate.UnknownDataException;
import org.diy.traffic.util.TLog;

public class DiyFluSink extends TrafficSink {

	private static String sinkType = "DIYFLU_DB_SINK";
	
	private static String mysqlDbName;
	private static String mysqlUsername;
	private static String mysqlPassword;

	private Connection dbConn = null;
	
	public DiyFluSink(int rateLimit) throws TrafficSinkException {
		super(sinkType, rateLimit);
		
		initDB();
	}

	public void initDB() throws TrafficSinkException {
		try {
			dbConn = DriverManager.getConnection(
					String.format("jdbc:mysql://localhost/%s?user=%s&password=%s", mysqlDbName, mysqlUsername, mysqlPassword)
					);
		} catch (SQLException e) {
			TLog.log(Level.WARNING, e);
			throw new TrafficSinkException(String.format("Can't connect to the database with dbname: '%s', username: '%s', password: '%s'",
					mysqlDbName, mysqlUsername, mysqlPassword));

		}
	}
	
	public static DiyFluSink createFromConfigData(HashMap<String, String> configParams) throws TrafficSinkException {
		mysqlDbName   = configParams.get("db");
		mysqlUsername = configParams.get("username");
		mysqlPassword = configParams.get("password");
		
		if ((mysqlDbName == null) || (mysqlUsername == null) || (mysqlPassword == null)) {
			TLog.log(Level.SEVERE, "DiyFluSink requires a 'db', 'username', and 'password' parameter for configuration! None provided.");
			return null;
		}
		
		return new DiyFluSink(5000);
	}
	
	@Override
	public boolean processUpdate(TrafficUpdate update)
			throws TrafficSinkException {

		Statement stmt = null;
		
		try {
			
			//System.out.println(String.format("****************Got a Flu Update: %s, %s, %s, %s", 
			//		update.getData("status_id"), update.getData("city_id"), update.getData("text"), update.getData("mentioner")));
		
			
			stmt = dbConn.createStatement();
			int query_result = stmt.executeUpdate(
					String.format("INSERT INTO mentions (mentioned_at, message, link, exact_location, phrase_id, city_id, mentioner, created_at, updated_at, source_id) "+
							"VALUES (\"%s\", \"%s\", \"%s\", \"%s, %s\", %d, %d, \"%s\", %s, %s, %s) ON DUPLICATE KEY UPDATE updated_at=updated_at",
							(String)update.getData("mentioned_at"),
							scrubStatusMessage((String) update.getData("text")),
							(String)update.getData("link"),
							(String)update.getData("city_name"),
							(String)update.getData("country_name"),
							Integer.parseInt((String)update.getData("phrase_id")),
							Integer.parseInt((String)update.getData("city_id")),
							(String)update.getData("mentioner"),
							"NOW()",
							"NOW()",
							Integer.parseInt((String) update.getData("status_id")))
					);
			//TLog.log(Level.FINE, String.format("DiyFluSink modified %d rows in the database", query_result));
			//System.out.println("Data successfully entered into db!!! :) :) :)");
		} catch (UnknownDataException e) {
			// TODO Auto-generated catch block
			TLog.log(Level.WARNING, e);
			throw new TrafficSinkException("DiyFluSource: Error while parsing data from update!");
		} catch (SQLException e) {
			// TODO Auto-generated catch block
			TLog.log(Level.WARNING, e);
			throw new TrafficSinkException("DiyFluSource: Error entering update into database!");
		} finally {
			try {
				if (stmt != null) stmt.close();
			} catch (SQLException e) {
				// TODO Auto-generated catch block
				TLog.log(Level.WARNING, e);
			}
		}
		
		update = null;
		System.gc();
		return true;
	}
	
	private String scrubStatusMessage(String statusMessage) {
		statusMessage = statusMessage.replace("\\", "\\\\");
		return statusMessage;
	}

}
